package test.top2.com.atguigu.app.fun;


import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import test.top2.com.atguigu.bean.TableProcess;
import test.top2.com.atguigu.common.GmallConfig;
import test.top2.com.atguigu.utils.DruidDSUtil;
import test.top2.com.atguigu.utils.JdbcUtil;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * ClassName: DIMTableProcessFunction
 * Package: test.top2.com.atguigu.app.fun
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/4/27 18:36
 * @Version 1.0
 */
public class DIMTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject>{

    private MapStateDescriptor<String, test.top2.com.atguigu.bean.TableProcess> stateDescriptor;

    private HashMap<String,TableProcess> tableProcessHashMap;

    public DIMTableProcessFunction(MapStateDescriptor<String, test.top2.com.atguigu.bean.TableProcess> stateDescriptor) {
        this.stateDescriptor = stateDescriptor;
    }


    //预加载  防止丢失数据
    @Override
    public void open(Configuration parameters) throws Exception {
        // 创建hashmap来存储状态
        tableProcessHashMap = new HashMap<>();

        //使用jdbc方式读取配置信息表，将数据存入hashmap
        Connection connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall_config?" + "user=root&password=000000&useUnicode=true&" + "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");

        List<TableProcess> queryList = JdbcUtil.queryList(connection,
                "select * from table_process where sink_type ='dim' ",
                TableProcess.class, true);

        //遍历放入hashmap集合
        for (TableProcess tableProcess : queryList) {
            //检验表（首次启动Phoenix中没有表，需要创建）
            checkTable(tableProcess);
            tableProcessHashMap.put(tableProcess.getSourceTable(),tableProcess);
        }

        //关闭连接
        connection.close();

    }

    //value:{"before":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1669162876406,"snapshot":"false","db":"gmall-220623-config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1669162876406,"transaction":null}
    //TODO 广播流
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

        //上下文获取广播状态
        BroadcastState<String,TableProcess> broadcastState =
                ctx.getBroadcastState(stateDescriptor);

        //0. 如果状态为d删除，则删除状态中的数据
        JSONObject jsonObject = JSONObject.parseObject(value);
        if ("d".equals(jsonObject.getString("op"))){
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("before"), TableProcess.class);
            String table = tableProcess.getSinkTable();
            broadcastState.remove(table);

            //删除hashmap中的值
            tableProcessHashMap.remove(table);

        }else{

            //1.将数据转为tableprocess对象
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("after"), TableProcess.class);

            //2.校验建表
            checkTable(tableProcess);

            //3.将数据写入状态
            broadcastState.put(tableProcess.getSourceTable(),tableProcess);

        }
    }

    //TODO 校验建表方法
    private void checkTable(TableProcess tableProcess) {

        //获取连接
        DruidPooledConnection phoenixConn =null;
        PreparedStatement preparedStatement = null;

        //获取相关信息
        String sinkTable = tableProcess.getSinkTable();
        String sinkColumns = tableProcess.getSinkColumns();
        String sinkPk = tableProcess.getSinkPk();
        String sinkExtend = tableProcess.getSinkExtend();

        try {
            phoenixConn = DruidDSUtil.getPhoenixConn();
            //拼接建表语句  create table if not exists db.tableName( id varchar primary key,name varchar,sex varchar)xxx


            if (sinkPk == null || "".equals(sinkPk)){
                sinkPk ="id";
            }
            if (sinkExtend == null){
                sinkExtend="";
            }

            StringBuilder createTableSql = new StringBuilder("create table if not exists " + GmallConfig.HBASE_SCHEMA + "." + sinkTable + "(");

            String[] columnArr = sinkColumns.split(",");

            for (int i = 0; i < columnArr.length; i++) {
                String column = columnArr[i];
                if (sinkPk.equals(column)){
                    createTableSql.append(column).append(" varchar primary key ");
                }else {
                    createTableSql.append(column).append(" varchar ");
                }
                if (i< columnArr.length -1){
                    createTableSql.append(",");
                }else{
                    createTableSql.append(")");
                }

            }

            //打印建表语句
            System.out.println("建表语句为："+createTableSql);

            //预编译sql语句
            preparedStatement = phoenixConn.prepareStatement(createTableSql.toString());

            //执行sql
            preparedStatement.execute();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("建表" + sinkTable + "失败!");
        } finally {
            //归还连接
            if (preparedStatement != null){
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

            if (phoenixConn != null){
                try {
                    phoenixConn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

        }

    }


    //Value:{"database":"gmall-221109-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null}}
    //TODO 主流
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {

        //1.获取状态信息
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(stateDescriptor);
        String table = value.getString("table");
        TableProcess tableProcess = broadcastState.get(table);

        //主流数据先到 从hashmap中也要获取
        TableProcess mapTableProcess = tableProcessHashMap.get(table);

        //insert update delete bootstrap-start bootstrap-insert bootstrap-complete
        String type = value.getString("type");

        //2.行过滤
        if ((tableProcess != null || mapTableProcess != null) && ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type))){

            //防止空指针
            if (table == null){
                tableProcess = mapTableProcess;
            }

            //列过滤
            filterColumns(value.getJSONObject("data"),tableProcess.getSinkColumns());

            //3.写出
            //多输出sink_table,方便输出phoenix
            value.put("sink_table", tableProcess.getSinkTable());
            out.collect(value);

        }else{
            System.out.println("该表:" + table + "没有对应的配置信息！或者类型不匹配:" + type);
        }

    }

    /**
     *
     * @param data  {"id":"1001","name":"zs","sex":"male","tm_name":"aaa"}
     * @param sinkColumns  id ,tm_name
     */
    //TODO 列过滤方法
    private void filterColumns(JSONObject data, String sinkColumns) {

        String[] columns = sinkColumns.split(",");
        List<String> columnsList = Arrays.asList(columns); //封装程集合  防止 包含关系 name  tm_name

//        Set<Map.Entry<String, Object>> entries = data.entrySet();
//        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
//        while (iterator.hasNext()){
//            Map.Entry<String, Object> next = iterator.next();
//            if ( !columnsList.contains(next.getKey())){
//                iterator.remove();
//            }
//        }

        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !columnsList.contains(next.getKey()));

    }
}
